package mylab.io.base;

import com.google.common.base.Preconditions;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import mylab.base.foundation.lang.Callback;
import mylab.io.base.context.DasContext;
import mylab.io.base.fsm.DasState;
import mylab.io.base.fsm.DasStatus;
import mylab.io.base.fsm.FsmContext;
import mylab.io.base.model.Msg;
import mylab.io.base.property.DasProperties;
import mylab.io.base.sync.Synchro;
import mylab.io.base.util.DasUtil;
import mylab.io.base.util.MsgUtil;

/**
 * 通用接入服务模板类，方便集成第三方开源通讯协议
 */
@Slf4j
@Getter
public abstract class BaseDas<C extends DasContext<P, M>, P extends DasProperties, M extends Msg>
        extends FsmContext
        implements Das<C, P, M> {

    protected C context;
    protected P properties;

    @Setter
    protected DasWatcher<M> watcher;

    public BaseDas(C context) {
        super();
        this.context = context;
        this.properties = context.getDasPropertites();
        this.id = this.properties.getId();

        initialize(null);
    }


    //------------------------------------ public  ----------------------------

    @Override
    public DasStatus getStatus() {
        return getDasState().getStatus();
    }

    @Override
    public void setDasState(DasState dasState) {
        super.setDasState(dasState);
        getContext().onDasStatusChanged(dasState.getStatus());
    }

    @Override
    public int getSyncTimeoutSeconds() {
        return this.getContext().getDasPropertites().getCommon().getSync().getTimeoutSeconds();
    }

    @Override
    public void asyncSend(M msg, Callback<Void> callback) {
        try {
            checkBeforeSend(msg);
            asyncSendFunc(msg, Callback.simple(callback, (Consumer<Void>) unused -> {
                DasUtil.logMsg(msg, true, false);
            }));

        } catch (Exception e) {
            Callback.dealOnException(callback, e);
        }
    }

    /**
     * 同步发送
     * 默认机制：通过收发消息的code配对表实现同步机制
     */
    @Override
    public M syncSend(M msg, long timeout, TimeUnit unit) throws Exception {
        checkBeforeSend(msg);

        Synchro synchro = context.getSynchro();
        Preconditions.checkNotNull(synchro, "synchro is null");

        //缓存请求
        synchro.cacheRequest(msg);

        //执行异步发送
        asyncSendFunc(msg, Callback.simple((Consumer<Void>) unused -> DasUtil.logMsg(msg, true, true)));

        SyncSendResult result = new SyncSendResult();
        CountDownLatch cd = new CountDownLatch(1);

        try {
            //等待回应
            Msg res = synchro.getResponse(msg, timeout, unit);
            DasUtil.logMsg(res, false, true);
            result.setResponse(res);
        } catch (Exception e) {
            result.setThrowable(e);
        } finally {
            cd.countDown();
            //无论正常收到回应还是请求超时，都要清理缓存
            synchro.cleanCache(msg);
        }

        //等待异步回应
        cd.await();
        //返回应答msg
        return result.checkResult();
    }

    //------------------------------------ protected  ----------------------------

    @Override
    protected void initializeFunc(Callback<Void> callback) {
        this.context.init(this);
        Callback.dealOnSuccess(callback);
    }

    @Override
    protected void disposeFunc(Callback<Void> callback) {
        this.context.dispose();
        Callback.dealOnSuccess(callback);
    }

    protected void checkBeforeSend(M msg) {
        Preconditions.checkState(getStatus().equals(DasStatus.Running), "Das is not running");
        Preconditions.checkState(MsgUtil.isValid(msg), "Invalid msg");
    }

    //------------------------------------ abstract method  ----------------------------

    /**
     * 建造通讯链路
     * 通讯链路的建立有同步与异步两种方式
     * 同步方式下，应以顺序代码的方式调用 onPipeRunningChanged(true)
     * 异步方式下，应在相应的回调方法内调用 onPipeRunningChanged(true)
     */
    protected abstract void startFunc(Callback<Void> callback);

    /**
     * 拆除通讯链路
     */
    protected abstract void stopFunc(Callback<Void> callback);

    /**
     * 异步发送
     */
    protected abstract void asyncSendFunc(M msg, Callback<Void> callback);
}
